package mn.idax.exchange.otc.service;

import com.alibaba.fastjson.JSON;
import mn.idax.exchange.otc.dto.ChatProtocol;
import mn.idax.exchange.otc.entity.ChatLog;
import org.apache.commons.lang.StringUtils;
import org.atmosphere.cpr.AtmosphereResource;
import org.atmosphere.cpr.Broadcaster;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.Collection;

/**
 * 系统消息
 * 由订单状态变更触发，通过redis publish
 */
@Service
public class ChatSubService implements MessageListener, CommandLineRunner {

    private static final Logger logger = LoggerFactory.getLogger(ChatRoomService.class);

    private static final int SYS_SENDER_ID = 0;

    private final StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();

    @Autowired
    private ChatLogService chatLogService;

    @Override
    @Async
    public void run(String... strings) {
        logger.info("----chat sub service is running----");
    }

    /**
     * sub message from redis
     *
     * @param message
     * @param bytes
     */
    @Override
    public void onMessage(Message message, byte[] bytes) {

        byte[] body = message.getBody();
        byte[] channel = message.getChannel();

        /**build **/
        String msg = stringRedisSerializer.deserialize(body);
        String topic = stringRedisSerializer.deserialize(channel);
        ChatProtocol chatProtocol = JSON.parseObject(msg, ChatProtocol.class);

        if (StringUtils.isEmpty(chatProtocol.getOrderId())) {
            logger.warn("----receive message from redis, but orderId not found----");
            return;
        }
        chatProtocol.setSenderId(SYS_SENDER_ID);

        /**save chatLog to mongo**/
        if (saveChatLogToMongo(chatProtocol) != null) {

            chatProtocol.setSaved(true);

            /**get broadcaster by chatRoom(orderId)**/
            Broadcaster broadcaster = ChatRoomService.broadcasterFactory.lookup(chatProtocol.getOrderId());
            if (broadcaster == null) {
                logger.warn("----receive message from redis, but broadcaster not found----");
                return;
            }

            /**check current connect**/
            Collection<AtmosphereResource> resourceCollection = broadcaster.getAtmosphereResources();
            if (resourceCollection.isEmpty()) {
                logger.info("----receive message from redis, but broadcaster not found----");
            } else {
                /**broadcast message **/
                broadcaster.broadcast(chatProtocol);
            }
        }
    }


    /**
     * saveChatLogToMongo
     *
     * @param chatProtocol
     * @return
     */
    private ChatLog saveChatLogToMongo(ChatProtocol chatProtocol) {
        try {
            if (chatProtocol != null) {
                ChatLog chatLog = new ChatLog();
                BeanUtils.copyProperties(chatProtocol, chatLog);
                logger.info("::::SaveChatLog::ChatSubService", JSON.toJSONString(chatLog));
                return chatLogService.insert(chatLog);
            }
        } catch (Exception e) {
            logger.error("----save chatLog to mongo error----", e);
        }
        return null;
    }
}
